/*
 * Copyright (C) 2014 Square, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
import { Call } from "../Call";
import { OkHttpClient } from "../OkHttpClient";
import { Request } from '../Request'
import { Response } from "../Response";
import { EventListener } from '../EventListener'
import { RealConnectionPool } from "./RealConnectionPool";
import { ExchangeFinder } from "./ExchangeFinder";
import { RealConnection } from "./RealConnection";
import { List } from "@kit.ArkTS";
import { Interceptor } from "../Interceptor";
import { RetryAndFollowUpInterceptor } from "../http/RetryAndFollowUpInterceptor";
import { BridgeInterceptor } from "../http/BridgeInterceptor";
import { CacheInterceptor } from "../cache/CacheInterceptor";
import { ConnectInterceptor } from "./ConnectInterceptor";
import { CallServerInterceptor } from "../http/CallServerInterceptor";
import { RealInterceptorChain } from "../http/RealInterceptorChain";
import { HttpUrl } from "../HttpUrl";
import { Address } from "../Address";
import { SSLSocketFactory } from "../Socket/SSLSocketFactory";
import { HostnameVerifier } from "../HostnameVerifier";
import { CertificatePinner } from "../CertificatePinner";
import { Exchange } from "./Exchange";
import { kotlinCheck } from "../utils/Utils";
import { IllegalArgumentException, IllegalStateException, InterruptedIOException, IOException } from "../Error/error";
import { socket } from "@kit.NetworkKit";
import { Runnable } from "../concurrent/Runnable";
import { Callback } from '../Callback'

/**
 * Bridge between OkHttp's application and network layers. This class exposes high-level application
 * layer primitives: connections, requests, responses, and streams.
 *
 * This class supports [asynchronous canceling][cancel]. This is intended to have the smallest
 * blast radius possible. If an HTTP/2 stream is active, canceling will cancel that stream but not
 * the other streams sharing its connection. But if the TLS handshake is still in progress then
 * canceling may break the entire connection.
 */

export class RealCall implements Call {

  client: OkHttpClient;
  originalRequest: Request;
  forWebSocket: boolean;
  connectionPool: RealConnectionPool;
  eventListener: EventListener;
  _timeout: () => void;

  constructor(client: OkHttpClient, originalRequest: Request, forWebSocket: boolean) {
    this.client = client;
    this.originalRequest = originalRequest;
    this.forWebSocket = forWebSocket;
    this.connectionPool = this.client.connectionPool.delegate
    this.eventListener = client.eventListenerFactory.create(this)
  }

  setCallTimeout(onCancel: () => void, timeoutMs: number): () => void {
    if (timeoutMs <= 0) {
      return () => {}; // 无需超时
    }

    const timer = setTimeout(() => {
      onCancel();
    }, timeoutMs);

    // 返回取消函数
    return () => {
      clearTimeout(timer);
    };
  }

  private executed = false // new Int32Array(new SharedArrayBuffer(4)) // 原子化操作,用于判断多线程请求是否已执行

  // These properties are only accessed by the thread executing the call.

  /** Initialized in [callStart]. */
  private callStackTrace: unknown | null = null

  /** Finds an exchange to send the next request and receive the next response. */
  private exchangeFinder: ExchangeFinder | null = null

  connection: RealConnection | null = null
  // private set

  private _timeoutEarlyExit = false
  interceptorScopedExchange: Exchange | null = null
  // private set
  /** True if this call still has a request body open. */
  private requestBodyOpen = false

  /** True if this call still has a response body open. */
  private responseBodyOpen = false

  /** True if there are more exchanges expected for this call. */
  private expectMoreExchanges = true

  private canceled = false    //@Volatile 多线程需要使用的变量
  private exchange?: Exchange = null    //@Volatile 多线程需要使用的变量
  connectionToCancel?: RealConnection = null    //@Volatile 多线程需要使用的变量
  timeout() {
    return this._timeout
  }

  // @SuppressWarnings("CloneDoesntCallSuperClone") // We are a final type & this saves clearing state.
  clone() {
    return new RealCall(this.client, this.originalRequest, this.forWebSocket)
  }

  request(): Request {
    return this.originalRequest
  }

  async cancel() {
    if (this.canceled) return // Already canceled.
    this.canceled = true
    this.exchange?.cancel()
    await this.connectionToCancel?.cancel()
    this.eventListener.canceled(this)
  }

  isCanceled(): boolean{
    return this.canceled
  }

  async execute(): Promise<Response>{ // 同步执行优先实现--单线程执行即可
    if (this.executed === false){
      this.executed = true
    }else{
      throw new IOException('Already Executed')
    }

    this._timeout = this.setCallTimeout(this.cancel,this.client.callTimeoutMillis); // 启动超时器

    this.callStart()
    try {
      this.client.dispatcher.executed(this)
      return this.getResponseWithInterceptorChain()
    } finally {
      this.client.dispatcher.finished(this) // 后续待修改
    }
  }

  enqueue(responseCallback: Callback) {  // 异步执行需多线程,暂不实现
    if (this.executed === false){
      this.executed = true
    }else{
      throw new IllegalStateException('Already Executed')
    }
    this.callStart()
    this.client.dispatcher.enqueue(new AsyncCall(responseCallback,  this))
  }

  isExecuted(): boolean{
    return this.executed
  }

  callStart() {
    // this.callStackTrace = Platform.get().getStackTraceForCloseable("response.body().close()")  // 获取当前堆栈信息
    this.eventListener.callStart(this)
  }

  // @Throws(IOException::class)
  async getResponseWithInterceptorChain(): Promise<Response> {
    // Build a full stack of interceptors.
    let interceptors = new List<Interceptor>()
    for (const interceptor of this.client.interceptors){
      interceptors.add(interceptor)
    }
    interceptors.add(new RetryAndFollowUpInterceptor(this.client))
    interceptors.add(new BridgeInterceptor(this.client.cookieJar))
    interceptors.add(new CacheInterceptor(this.client.cache))
    interceptors.add(new ConnectInterceptor())
    if (!this.forWebSocket) {
      for (const interceptor of this.client.networkInterceptors){
        interceptors.add(interceptor)
      }
    }
    interceptors.add(new CallServerInterceptor(this.forWebSocket))

    const chain = new RealInterceptorChain(
      this,
      interceptors,
      0,
      undefined,
      this.originalRequest,
      this.client.connectTimeoutMillis,
      this.client.readTimeoutMillis,
      this.client.writeTimeoutMillis
    )

    let calledNoMoreExchanges = false
    try {
      let response = await chain.proceed(this.originalRequest)
      if (this.isCanceled()) {
        // response.closeQuietly()
        throw new IOException('Canceled')
      }
      return response
    } catch (e) {
      calledNoMoreExchanges = true
      throw await this.noMoreExchanges(e)
    } finally {
      // 注释删除:  需要对call和connection进行松绑
      if (!calledNoMoreExchanges) {
        await this.noMoreExchanges(null)
      }
    }
  }

  /**
   * Prepare for a potential trip through all of this call's network interceptors. This prepares to
   * find an exchange to carry the request.
   *
   * Note that an exchange will not be needed if the request is satisfied by the cache.
   *
   * @param newExchangeFinder true if this is not a retry and new routing can be performed.
   */
  enterNetworkInterceptorExchange(request: Request, newExchangeFinder: boolean) {
    kotlinCheck(this.interceptorScopedExchange === null, '')

    // synchronized(this)
    //  {
    kotlinCheck(!this.responseBodyOpen,
      "cannot make a new request because the previous response is still open: " +
        "please call response.close()")
    kotlinCheck(!this.requestBodyOpen, '')
    // }
    if (newExchangeFinder) {
      this.exchangeFinder = new ExchangeFinder(
        this.connectionPool,
        this.createAddress(request.url),
        this,
        this.eventListener
      )
    }
  }

  /** Finds a new or pooled connection to carry a forthcoming request and response. */
  async initExchange(chain: RealInterceptorChain): Promise<Exchange> {
    // synchronized(this)
    // {
    kotlinCheck(this.expectMoreExchanges, "released")
    kotlinCheck(!this.responseBodyOpen, '')
    kotlinCheck(!this.requestBodyOpen, '')

    let exchangeFinder = this.exchangeFinder
    let codec = await exchangeFinder.find(this.client, chain)
    let result = new Exchange(this, this.eventListener, exchangeFinder, codec)
    this.interceptorScopedExchange = result
    this.exchange = result
    // synchronized(this)
    // {
      this.requestBodyOpen = true
      this.responseBodyOpen = true
    // }

    if (this.canceled) throw new IOException("Canceled")
    return result
  }

  acquireConnectionNoEvents(connection: RealConnection) {
    // connection.assertThreadHoldsLock()
    kotlinCheck(this.connection === null, '')

    this.connection = connection
    this.connection.calls.add(this) // new CallReference(this, callStackTrace)
  }


  /**
   * Releases resources held with the request or response of [exchange]. This should be called when
   * the request completes normally or when it fails due to an exception, in which case [e] should
   * be non-null.
   *
   * If the exchange was canceled or timed out, this will wrap [e] in an exception that provides
   * that additional context. Otherwise [e] is returned as-is.
   */
  async messageDone(exchange: Exchange, requestDone: boolean, responseDone: boolean, e: Error | null): Promise<Error> | null {
    if (exchange !== this.exchange) return e // This exchange was detached violently!

    let bothStreamsDone = false
    let callDone = false
    // synchronized(this) {
      if (requestDone && this.requestBodyOpen || responseDone && this.responseBodyOpen) {
        if (requestDone) this.requestBodyOpen = false
        if (responseDone) this.responseBodyOpen = false
        bothStreamsDone = !this.requestBodyOpen && !this.responseBodyOpen
        callDone = !this.requestBodyOpen && !this.responseBodyOpen && !this.expectMoreExchanges
      }
    // }

    if (bothStreamsDone) {
      this.exchange = null
      this.connection?.incrementSuccessCount()
    }
    console.log("okhttp test log :: " , "messageDone : 开始")

    if (callDone) {
      return await this.callDone(e)
    }

    return e
  }

  async noMoreExchanges(e?: Error): Promise<Error> {
    let callDone = false
    // synchronized(this) {
    if (this.expectMoreExchanges) {
      this.expectMoreExchanges = false
      callDone = !this.requestBodyOpen && !this.responseBodyOpen
    }
    // }
    if (callDone) {
      return await this.callDone(e)
    }
    return e
  }

  async callDone(e?: Error): Promise<Error> {
    console.log("okhttp test log :: " , "callDone : 开始")

    // assertThreadDoesntHoldLock()
    let connection = this.connection
    if (connection !== null && connection !== undefined) {
      // connection.assertThreadDoesntHoldLock()
      let socket =
        // synchronized(connection) {
        this.releaseConnectionNoEvents() // Sets this.connection to null.
      // }
      console.log("okhttp test log :: " , "callDone socket : ", socket === null)
      if (this.connection === null || this.connection === undefined) {
        console.log("okhttp test log :: " , "callDone socket : close")
        await socket?.close()
        this.eventListener.connectionReleased(this, connection)
      } else {
        kotlinCheck(socket === null || socket === undefined,"") // If we still have a connection we shouldn't be closing any sockets.
      }
    }

    let result = this.timeoutExit(e)
    if (e !== null && e !== undefined) {
      this.eventListener.callFailed(this, result)
    } else {
      this.eventListener.callEnd(this)
    }
    return result
  }

  /**
   * Remove this call from the connection's list of allocations. Returns a socket that the caller
   * should close.
   */
  releaseConnectionNoEvents(): socket.TCPSocket | socket.TLSSocket | null {
    let connection = this.connection
    // connection.assertThreadHoldsLock()

    let calls = connection.calls
    let index: number = calls.getIndexOf(this)
    kotlinCheck(index != -1,"")
    calls.removeByIndex(index)
    this.connection = null

    if (calls.isEmpty()) {
      connection.idleAtNs = Date.now() * 1000000 // 转换为纳秒
      if (this.connectionPool.connectionBecameIdle(connection)) {
        console.log("okhttp test log :: " , "releaseConnectionNoEvents connectionBecameIdle : true")
        return connection.socket
      }
    }
    return null
  }

  private timeoutExit(cause: Error): Error { // <E : IOException?>
    console.log("okhttp test log :: timeoutExit input cause:", cause?.name, cause?.message)
    if (this._timeoutEarlyExit) return cause
    // if (!this.timeout.exit()) return cause
    // 只有在没有原始错误时才返回超时异常
    let e = new InterruptedIOException("timeout")
    if (cause !== null && cause !== undefined) {
      e.initCause(cause)
    }
    return e
  }

  /**
   * Stops applying the timeout before the call is entirely complete. This is used for WebSockets
   * and duplex calls where the timeout only applies to the initial setup.
   */
  timeoutEarlyExit() {
    kotlinCheck(!this._timeoutEarlyExit,"")
    this._timeoutEarlyExit = true
    this._timeout() // 关闭定时器
  }

  /**
   * @param closeExchange true if the current exchange should be closed because it will not be used.
   *     This is usually due to either an exception or a retry.
   */
  exitNetworkInterceptorExchange(closeExchange: boolean) {
    // synchronized(this) {
    kotlinCheck(this.expectMoreExchanges,"released")
    // }

    if (closeExchange) {
      this.exchange?.detachWithViolence()
    }

    this.interceptorScopedExchange = null
  }

  private createAddress(url: HttpUrl): Address {
    let sslSocketFactory: SSLSocketFactory = null
    let hostnameVerifier: HostnameVerifier = null
    let certificatePinner: CertificatePinner = null
    if (url.isHttps) {
      sslSocketFactory = this.client.sslSocketFactory
      hostnameVerifier = this.client.hostnameVerifier
      certificatePinner = this.client.certificatePinner
    }

    return new Address(
      url.host,
      url.port,
      this.client.dns,
      this.client.socketFactory,
      sslSocketFactory,
      hostnameVerifier,
      certificatePinner,
      this.client.proxyAuthenticator,
      this.client.proxy,
      this.client.protocols,
      this.client.connectionSpecs,
      this.client.proxySelector
    )
  }

  retryAfterFailure() {
    return this.exchangeFinder!!.retryAfterFailure()
  }

  /**
   * Returns a string that describes this call. Doesn't include a full URL as that might contain
   * sensitive information.
   */
  toLoggableString(): string {
    let res = ''
    if (this.isCanceled()){
      res = 'canceled '
    }
    if (this.forWebSocket) {
      res = 'web socket'
    }else{
      res = 'call'
    }
    return res += ' to ' + this.redactedUrl()
  }

  redactedUrl(): string {
    return this.originalRequest.url.redact()
  }
}

export class AsyncCall implements Runnable {
  private responseCallback: Callback
  private realCall: RealCall

  constructor(responseCallback: Callback, realCall: RealCall) {
    this.responseCallback = responseCallback
    this.realCall = realCall
  }

  // @Volatile var callsPerHost = AtomicInteger(0)
  callsPerHost = 0
  // private set

  reuseCallsPerHostFrom(other: AsyncCall) {
    this.callsPerHost = other.callsPerHost
  }

  _host: string

  get host() {
    return this.realCall.originalRequest.url.host
  }

  _request: Request

  get request() {
    return this.realCall.originalRequest
  }

  _call: RealCall

  get call() {
    return this.realCall
  }

  /**
   * Attempt to enqueue this async call on [executorService]. This will attempt to clean up
   * if the executor has been shut down by reporting the call as failed.
   */
  executeOn(){ //executorService: ExecutorService) {
    // this.realCall.client.dispatcher.assertThreadDoesntHoldLock()

    var success = false
    try {
      // executorService.execute(this)
      this.run()
      success = true
    } catch (e) {
      let ioException = new InterruptedIOException("executor rejected")
      // ioException.initCause(e)
      this.realCall.noMoreExchanges(ioException)
      this.responseCallback.onFailure(this.realCall, ioException)
    } finally {
      if (!success) {
        this.realCall.client.dispatcher.finishedAsync(this) // This call is no longer running!
      }
    }
  }

  async run() {
    // threadName("OkHttp ${redactedUrl()}") {
    let signalledCallback = false
    this.realCall._timeout = this.realCall.setCallTimeout(this.realCall.cancel, this.realCall.client.callTimeoutMillis)
    try {
      let response = await this.realCall.getResponseWithInterceptorChain()
      signalledCallback = true
      this.responseCallback.onResponse(this.realCall, response)
    } catch (e) {
      if (e instanceof IOException) {
        if (signalledCallback) {
          // Do not signal the callback twice!
          console.log("Callback failure for " + this.call.toLoggableString(), e)
        } else {
          this.responseCallback.onFailure(this.realCall, e)
        }
      } else {
        await this.realCall.cancel()
        if (!signalledCallback) {
          let canceledException = new IOException("canceled due to $t")
          canceledException.addSuppressed(e)
          this.responseCallback.onFailure(this.realCall, canceledException)
        }
        throw e
      }
    } finally {
      this.realCall.client.dispatcher.finishedAsync(this)
    }
    // }
  }
}